package com.csw.spark

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo09ReduceByKey {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setAppName("reduceByKey")
      .setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val rdd1: RDD[String] = sc.textFile("spark/data/students.txt")

    //取出每一个(班级,1)
    val rdd2: RDD[(String, Int)] = rdd1.map(i => {
      val clazz: String = i.split(",")(4)
      (clazz, 1)
    })

    /**
      * 统计班级人数
      */
    //先分组
    val rdd3: RDD[(String, Iterable[Int])] = rdd2.groupByKey()

    //    rdd3.foreach(println)

    //输出(班级,班级人数)
    val rdd4: RDD[(String, Int)] = rdd3.map(kv => {
      val clazz: String = kv._1

      //班级的人数
      val count: Int = kv._2.sum

      (clazz, count)
    })

    //    rdd4.foreach(println)

    /**
      * reduceByKey：通过key对value进行聚合，一般用于加和计算，需要传入一个聚合函数
      * 效率比groupByKey高，reduceByKey可以在map端进行预聚合， (可以减少shuffle过程中传输的数据量)
      *
      * reduceByKey只能用于聚合操作，功能比groupByKey弱，性能更高
      *
      */

    val rdd5: RDD[(String, Int)] = rdd2.reduceByKey((x, y) => x + y)

//    rdd5.foreach(println)

    //简写，函数的参数只使用了一次，可以通过下划线代替
    val rdd6: RDD[(String, Int)] = rdd2.reduceByKey(_ + _)

    rdd6.foreach(println)
  }
}
